package com.sh.promise;


import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.sh.promise.then.OnCatchedExecutor;
import com.sh.promise.then.OnCompleteListener;
import com.sh.promise.then.OnFulfilledExecutor;
import com.sh.promise.then.OnRejectedExecutor;
import com.sh.promise.util.PromiseCountDownLatch;
import com.sh.promise.util.PromiseUtil;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.*;

public class Promise extends AbstractPromise{

    private PromiseHandler promiseHandler;
    private PromiseExecutor promiseExecutor;

    private Set<OnCompleteListener> onCompleteListenerSet = new HashSet<>();

    private ExecutorService threadPool;

    private Object externalInput;
    private Object promiseInput;

    private Future future;

    private int stateOffset = -1;

    private Promise(){}
    private Future startThread(Callable callable){
        if(this.threadPool !=null){
            this.future = this.threadPool.submit(callable);
        }else{
            ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("promise-thread-%d").build();
            FutureTask futureTask = new FutureTask<Object>(callable);
            namedThreadFactory.newThread(futureTask).start();
            this.future = futureTask;
        }
        return this.future;
    }


    /**
     * 启动线程
     */
    private void executePromise(IPromise otherPromise){
        if(otherPromise!=null){
            //如果otherPromise处于等待态，promise需保持为等待态直至otherPromise被执行或拒绝
            if(otherPromise.getStatus().equals(Status.PENDING)){
                otherPromise.then(null);
            }
            if(otherPromise.getStatus().equals(Status.FULFILLED)){
                this.promiseInput = otherPromise.getResolvedData();
                startThread(this);
            }else{
                this.transformStatus(Status.REJECTED,otherPromise.getRejectedData());
            }
        }else{
            startThread(this);
        }
    }
    @Override
     public Object call(){
        if(this.getStatus().equals(Status.PENDING)&&null!=this.promiseHandler){
            try{
                //执行具体的方法
                final Object result = this.promiseHandler.run(this.promiseExecutor);
                //如果是直接返回数据，没有调用resolve，转换状态
                if(this.getStatus().equals(Status.PENDING)){
                    this.transformStatus(Status.FULFILLED,result);
                }
            }catch (Exception e){
                if(this.getStatus().equals(Status.PENDING)){
                    this.transformStatus(Status.REJECTED,e);
                }
            }finally {
                // this.isFinished = true;
            }
        }

        //调用监听
        if(this.onCompleteListenerSet !=null&&this.onCompleteListenerSet.size()>0){
            this.onCompleteListenerSet.forEach(listener->listener.listen(this.getResolvedData(),this.getRejectedData()));
        }
        return this.getStatus().equals(Status.FULFILLED)?this.getResolvedData():this.getRejectedData();
    }
    @Override
    public IPromise then(OnFulfilledExecutor onFulfilledExecutor){
        return then(onFulfilledExecutor,null);
    }
    @Override
    public IPromise then(OnFulfilledExecutor onFulfilledExecutor, OnRejectedExecutor onRejectedExecutor){
        try {
            if(this.getStatus().equals(Status.PENDING)){
                this.future.get();
            }
            if(this.getStatus().equals(Status.FULFILLED)){
                if(onFulfilledExecutor!=null){
                    //调用回调
                    Object r = onFulfilledExecutor.onFulfilled(this.getResolvedData());
                    if(r instanceof IPromise){
                        return (IPromise)r;
                    }else{
                        return finalPromise(r,true);
                    }
                }else{
                    return finalPromise(this.getResolvedData(),true);
                }
            }else{
                if(onRejectedExecutor != null){
                    onRejectedExecutor.onRejected(this.getRejectedData());
                }
                return finalPromise(this.getRejectedData(),false);
            }
        }catch (Exception e){
            return finalPromise(e,false);
        }
    }

    @Override
    public Future getFuture(){
        return this.future;
    }

    @Override
    public boolean cancel(){
        return this.future!=null&&this.future.cancel(true);
    }
    @Override
    public Object get(){
        try {
            if(this.getStatus().equals(Status.PENDING)){
                this.future.get();
            }
            if(stateOffset == 0){
                return this.getRejectedData();
            }
            if(this.stateOffset == 1 || this.getStatus().equals(Status.FULFILLED) ){
                return this.getResolvedData();
            }else{
                return this.getRejectedData();
            }
        }catch (Exception e){
            return new RuntimeException(e.getMessage());
        }
    }

    @Override
    public void setSuccess(Object obj){
       //this.setResolvedData(obj);
        this.stateOffset = 1;
        set(obj);
    }

    @Override
    public void setFailure(Throwable t) {
        this.stateOffset = 0;
        this.setException(t);
    }
    @Override
    public void addListener(OnCompleteListener onCompleteListener){
        //如果是非pending状态，立即调用监听
        if(this.getStatus().equals(Status.FULFILLED)){
            onCompleteListener.listen(this.getResolvedData(),null);
        }else if(this.getStatus().equals(Status.REJECTED)){
            onCompleteListener.listen(this.getResolvedData(),this.getRejectedData());
        }else{
            //pending状态，将监听加入集合
            this.onCompleteListenerSet.add(onCompleteListener);
        }
    }
    @Override
    public void pFinally(OnCompleteListener onCompleteListener){
        addListener(onCompleteListener);
    }
    @Override
    public IPromise pCatch(OnCatchedExecutor onCatchedExecutor){
        try {
            if(this.getStatus().equals(Status.PENDING)){
                this.future.get();
            }
            if(this.getStatus().equals(Status.REJECTED)){
                if(onCatchedExecutor!=null){
                    Object r = onCatchedExecutor.onCatched(this.getRejectedData());
                    if(r instanceof IPromise){
                        return (IPromise)r;
                    }else{
                        return finalPromise(r,true);
                    }
                }
                return finalPromise(this.getRejectedData(),false);
            }
            return finalPromise(this.getResolvedData(),true);
        }catch (Exception e){
            return finalPromise(e,false);
        }
    }

    /**
     * 创建一个不可变的promise
     * @param data
     * @param fulfilled
     * @return
     */
    private IPromise finalPromise(Object data,boolean fulfilled){
        return new Builder().finalPromise(data,fulfilled).build();
    }

    protected Object getExternalInput(){
        return this.externalInput;
    }
    protected Object getPromiseInput(){
        return this.promiseInput;
    }


    /**
     * 创建一个固定容量的线程池
     * @param size
     * @return
     */
    public static ExecutorService pool(int size){
        return pool(size,size);
    }

    /**
     * 创建一个线程池
     * @param minSize 最小线程数
     * @param maxSize 最大线程数
     * @return
     */
    public static ExecutorService pool(int minSize,int maxSize){
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("promise-pool-%d").build();
        ExecutorService fixedThreadPool = new ThreadPoolExecutor(minSize,maxSize,
                10L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(),namedThreadFactory);
        return fixedThreadPool;
    }
    public static class Builder{
        private Promise promise = new Promise();
        private IPromise otherPromise;
        public Builder(){
        }

        /**
         * 指定一个线程池用于执行promise任务，如果不指定，每一个promise都将启动一个线程
         * @param threadPool
         * @return
         */
        public Builder pool(ExecutorService threadPool){
            if(threadPool!=null){
                this.promise.threadPool = threadPool;
            }
            return this;
        }

        /**
         * 向Promise注入一个外部参数，可以在指定PromiseHandler时通过PromiseExecutor.getExternalInput()获取
         * @param externalInput
         * @return
         */
        public Builder externalInput(Object externalInput){
            this.promise.externalInput = externalInput;
            return this;
        }

        /**
         * 指定promise处理器
         * @param promiseHandler
         * @return
         */
        public Builder promiseHandler(PromiseHandler promiseHandler){
            this.promise.promiseHandler = promiseHandler;
            this.promise.promiseExecutor = new PromiseExecutor(this.promise);
            return this;
        }

        /**
         * 创建一个不可变的promise
         * @param finalData
         * @param fulfilled
         * @return
         */
        public Builder finalPromise(final Object finalData,boolean fulfilled){
            if(fulfilled){
                this.promise.transformStatus(Status.FULFILLED,finalData);
            }else{
                this.promise.transformStatus(Status.REJECTED,finalData);
            }
            return this;
        }

        /**传入一个promise
         * @param promise
         * @return
         */
        public Builder promise(IPromise promise){
            this.otherPromise = promise;
            return this;
        }
        public IPromise build(){
            if(this.promise.promiseHandler==null){
                this.promise.transformStatus(Status.FULFILLED,null);
            }else if(this.promise.getStatus().equals(Status.PENDING)){
                this.promise.executePromise(this.otherPromise);
            }
            return this.promise;
        }
    }
    public void set(Object obj){
        do{
            this.setResolvedData(obj);
        }while (!this.future.isDone());
    }

    public void setException(Throwable e){
        do{
            this.setRejectedData(e);
        }while (!this.future.isDone());
        /*do{
            this.setRejectedData(e);
        }while (!this.isFinished);*/
    }

    /**
     * @return IPromise
     */
    public static IPromise all(final IPromise ...promises){
        return all(null,promises);
    }

    /**
     * @param threadPool 线程池,指定新的实例 p的运行环境，为null时会为实例 p开启一个新线程。
     *                   线程池的容量最好比promises长度大1，避免实例p处于队列等待
     * @param promises promise数组
     * @return IPromise
     */
    public static IPromise all(ExecutorService threadPool,final IPromise ...promises){
        if(promises==null){
            return new Promise.Builder().finalPromise(null,true).build();
        }
        if(promises.length==0){
            return new Promise.Builder().finalPromise(new Object[0],true).build();
        }
        IPromise rejectedPromise = PromiseUtil.getFirstRejected(promises);
        if(rejectedPromise!=null){
            PromiseUtil.cancel(promises);
            return new Promise.Builder().finalPromise(rejectedPromise.getRejectedData(),false).build();
        }
        return new Promise.Builder().promiseHandler(handler->{
            final PromiseCountDownLatch counter = new PromiseCountDownLatch(promises.length);
            for(IPromise p:promises){
                p.addListener((resolvedData, throwable) -> {
                    if(throwable!=null){
                        counter.countDownAll();
                    }else{
                        counter.countDown();
                    }
                });
            }
            counter.await();
            IPromise rejectedPromise2 = PromiseUtil.getFirstRejected(promises);
            if(rejectedPromise2!=null){
                PromiseUtil.cancel(promises);
                handler.reject(rejectedPromise2.getRejectedData());
            }else{
                final Object[] datas = new Object[promises.length];
                for(int i=0;i<promises.length;i++){
                    IPromise p = promises[i];
                    datas[i] = p.getResolvedData();
                }
                handler.resolve(datas);
            }
            return null;
        }).pool(threadPool).build();
    }

    /**
     * * @return
     */
    public static IPromise race(IPromise ...promises){
        return race(null,promises);
    }
    /**
     * @param threadPool 线程池,指定新的实例 p的运行环境，为null时会为实例 p开启一个新线程。
     *                   线程池的容量最好比promises长度大1，避免实例p处于队列等待
     * @param promises
     * @return
     */
    public static IPromise race(ExecutorService threadPool,IPromise ...promises){
        if(promises==null){
            return new Promise.Builder().finalPromise(null,true).build();
        }
        if(promises.length==0){
            return new Promise.Builder().finalPromise(new Object[0],true).build();
        }
        IPromise finalPromise = PromiseUtil.getFinal(promises);
        if(finalPromise!=null){
            PromiseUtil.cancel(promises);
            return PromiseUtil.cloneFinal(finalPromise);
        }
        return new Promise.Builder().promiseHandler(handler->{
            final PromiseCountDownLatch counter = new PromiseCountDownLatch(promises.length);
            for(IPromise p:promises){
                p.addListener((resolvedData, throwable) -> counter.countDownAll());
            }
            counter.await();
            IPromise finalPromise2 = PromiseUtil.getFinal(promises);
            PromiseUtil.cancel(promises);
            if(Status.FULFILLED.equals(finalPromise2.getStatus())){
                handler.resolve(finalPromise2.getResolvedData());
            }else{
                handler.reject(finalPromise2.getRejectedData());
            }
            return null;
        }).pool(threadPool).build();
    }
}
